package com.desheng.bigdata.flink.time

import java.net.URL

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
  *     分配水印有两种方式：
  *         AssignerWithPeriodicWatermarks  ---->周期性的为一条记录设置一个水印时间，默认时间是200ms
  *         AssignerWithPunctuatedWatermarks ---非周期性，就位每一条记录设置一个水印的时间
  */
object _02TimeCharacteristicApp {
    def main(args: Array[String]): Unit = {

        val env = StreamExecutionEnvironment.getExecutionEnvironment

        //要基于eventtime进行处理
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

        val lines = env.fromCollection(List(
            UserClick(1, "张三", "http://bd.ujiuye.com", 1600239353208L),
            UserClick(2, "李四", "http://java.ujiuye.com", 1600239353608L),
            UserClick(3, "王五", "http://bd.ujiuye.com", 1600239353428L),
            UserClick(4, "赵六", "http://ui.ujiuye.com", 1600239353701L),
            UserClick(5, "周七", "http://python.ujiuye.com", 1600239352258L)
        ))

        lines.assignTimestampsAndWatermarks(
            new MyWatermarkAssigner()
        ).map(uc => {
            val url = new URL(uc.url)
            url.getHost
            (url.getHost, uc)
        }).keyBy(kv => kv._1)
            .timeWindow(Time.seconds(4))
            .process(new ProcessWindowFunction[(String, UserClick), (String, Int), String, TimeWindow] {
                override def process(key: String, context: Context,
                                     elements: Iterable[(String, UserClick)],
                                     out: Collector[(String, Int)]): Unit = {
                    elements.foreach{case (host, ucs) => {
                        println(ucs)
                    }}
                    out.collect((key, elements.size))
                }
            })
            .print()

        env.execute()
    }
}
class MyWatermarkAssigner extends AssignerWithPunctuatedWatermarks[UserClick] {

    private var currentMaxTimestamp: Long = 0L
    /**
      * 提取事件中的时间戳
      * @param uc  当前处理的元素
      * @param previousElementTimestamp
      * @return
      */
    override def extractTimestamp(uc: UserClick, previousElementTimestamp: Long): Long = {
        val timestamp = uc.timestamp + 2000L
        if (timestamp > currentMaxTimestamp) {
            currentMaxTimestamp = timestamp
        }
        timestamp
    }

    /**
      * 检测并获取下一次的水印，是在extractTimestamp方法之后被执行
      * @param uc
      * @param extractedTimestamp
      * @return
      */
    override def checkAndGetNextWatermark(uc: UserClick, extractedTimestamp: Long): Watermark = {
        if(uc.timestamp <= (currentMaxTimestamp - 2000L)) {//当前的事件时间在最大事件时间-t的范围内，那么认为所有的数据都已经抵达了
            new Watermark(extractedTimestamp)
        } else {
            null
        }
    }
}